#include "tcp_connection.h"
#include "callbacks.h"
#include "channel.h"
#include "logger.h"
#include "socket.h"
#include "timestamp.h"

#include <asm-generic/socket.h>
#include <cerrno>
#include <cstddef>
#include <errno.h>
#include <functional>
#include <netinet/tcp.h>
#include <strings.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <unistd.h>

static EventLoop *check_loop_not_null(EventLoop *loop) {
    if (loop == nullptr) {
        LOG_FATAL("%s:%s:%d main_loop is null! \n", __FILE__, __FUNCTION__, __LITTLE_ENDIAN);
    }
    return loop;
}

TcpConnection::TcpConnection(EventLoop *loop, const std::string &name_arg, int sockfd,
                             const InetAddress &local_addr, const InetAddress &peer_addr)
    : loop_(check_loop_not_null(loop)), name_(name_arg), state_(kConnecting), reading_(true),
      socket_(new Socket(sockfd)), channel_(new Channel(loop, sockfd)), local_addr_(local_addr),
      peer_addr_(peer_addr), high_water_mark_(64 * 1024 * 1024) // 64M
{
    channel_->set_read_callback(
        std::bind(&TcpConnection::handle_read, this, std::placeholders::_1));
    channel_->set_write_callback(std::bind(&TcpConnection::handle_write, this));
    channel_->set_close_callback(std::bind(&TcpConnection::handle_close, this));
    channel_->set_error_callback(std::bind(&TcpConnection::handle_error, this));

    LOG_INFO("TcpConnection::ctor[%s] at fd= %d\n", name_.c_str(), sockfd);
    socket_->set_keep_alive(true);
}

TcpConnection::~TcpConnection() {
    LOG_INFO("TcpConnection::dtor[%s] at fd=%d state=%d \n", name_.c_str(), channel_->fd(),
             (int)state_);
}

void TcpConnection::handle_read(Timestamp receive_time) {
    int saved_errno = 0;
    ssize_t n = input_buffer_.read_fd(channel_->fd(), &saved_errno);
    if (n > 0) {
        // 以建立连续的用户，有可读时间发生了，调用用户传入的回调操作on_message
        message_callback_(shared_from_this(), &input_buffer_, receive_time);
    } else if (n == 0) {
        handle_close();
    } else {
        errno = saved_errno;
        LOG_ERROR("TcpConnection::handle_read");
        handle_error();
    }
}

void TcpConnection::handle_write() {
    if (channel_->is_writing()) {
        int saved_errno = 0;
        ssize_t n = output_buffer_.write_fd(channel_->fd(), &saved_errno);
        if (n > 0) {
            output_buffer_.retrieve(n);
            if (output_buffer_.readable_bytes() == 0) {
                channel_->disable_writing();
                if (write_complete_callback_) {
                    // 唤醒loop_对应的thread线程，执行回调
                    loop_->queue_in_loop(std::bind(write_complete_callback_, shared_from_this()));
                }
                if (state_ == kDisconnecting) {
                    shutdown_in_loop();
                }
            }
        } else {
            LOG_ERROR("TcpConnection::handle_write");
        }
    } else {
        LOG_ERROR("TcpConnection::handle_write TcpConnection fd=%d is down ,no more writing \n",
                  channel_->fd());
    }
}

void TcpConnection::handle_close() {
    LOG_INFO("TcpConnection::handle_close fd = %d state=%d \n", channel_->fd(),
             atomic_load(&state_)); // 这里转化一下
    set_state(kDisconnected);
    channel_->disable_all();

    TcpConnectionPtr conn_ptr(shared_from_this());
    connection_callback_(conn_ptr); // 执行连接关闭的回调
    close_callback_(conn_ptr);      // 关闭连接的回调
}

void TcpConnection::handle_error() {
    int optval;
    socklen_t optlen = sizeof optval;
    int err = 0;
    if (::getsockopt(channel_->fd(), SOL_SOCKET, SO_ERROR, &optval, &optlen) < 0) {
        err = errno;
    } else {
        err = optval;
    }
    LOG_ERROR("TcpConnection::handle_error name:%s - SO_ERROR:%d \n", name_.c_str(), err);
}

void TcpConnection::send(const std::string &buf) {
    if (state_ == kConnected) {
        if (loop_->is_in_loop_thread()) {
            send_in_loop(buf.c_str(), buf.size());
        } else {
            loop_->run_in_loop(
                std::bind(&TcpConnection::send_in_loop, this, buf.c_str(), buf.size()));
        }
    }
}

void TcpConnection::send_in_loop(const void *data, size_t len) {
    ssize_t nwrote = 0;
    size_t remaining = len;
    bool fault_error = false;

    // 之前调用过该connection的shutdown，不能在进行发送了
    if (state_ == kDisconnected) {
        LOG_ERROR("disconnected,give up writing!");
        return;
    }

    // 表示channel_第一次开始写数据，而且缓冲区没有待发送数据
    if (!channel_->is_writing() && output_buffer_.readable_bytes() == 0) {
        nwrote = ::write(channel_->fd(), data, len);
        if (nwrote >= 0) {
            remaining = len - nwrote;
            if (remaining == 0 && write_complete_callback_) {
                // 既然在这里数据全部发送完成，就不用再给channel设置epollout事件了
                loop_->queue_in_loop(std::bind(write_complete_callback_, shared_from_this()));
            }
        } else // nwrote <0
        {
            nwrote = 0;
            if (errno != EWOULDBLOCK) {
                LOG_ERROR("TcpConnection::send_in_loop");
                if (errno == EPIPE || errno == ECONNRESET) // SIGPIPE RESET
                {
                    fault_error = true;
                }
            }
        }
    }

    // 说明当前这一次write，并没有把数据全部发出去，剩余数据需要放入缓冲区中，然后给channel
    // 注册epollout事件，poller发现tcp的发送缓冲区有空间，会通知相应的sock-channel,调用write_callback回调方法
    // 也就是调用TcpConnection::handle_write方法，把发送缓冲去的数据全部发送完成
    if (!fault_error && remaining > 0) {
        // 当前发送缓冲区剩余的待发送数据的长度
        size_t old_len = output_buffer_.readable_bytes();
        if (old_len + remaining >= high_water_mark_ && old_len < high_water_mark_ &&
            high_water_mark_callback_) {
            loop_->queue_in_loop(
                std::bind(high_water_mark_callback_, shared_from_this(), old_len + remaining));
        }
        output_buffer_.append((char *)data + nwrote, remaining);
        if (!channel_->is_writing()) {
            channel_
                ->enable_writing(); // 这里一定要注册channel的写事件，否则poller不会给channel通直epoller
        }
    }
}

// 关闭连接
void TcpConnection::shutdown() {
    if (state_ == kConnected) {
        set_state(kDisconnecting);
        loop_->run_in_loop(std::bind(&TcpConnection::shutdown_in_loop, this));
    }
}

void TcpConnection::shutdown_in_loop() {
    if (!channel_->is_writing()) // 说明output_buffer的数据全部发送
    {
        socket_->shutdown_write(); // 关闭写
    }
}

void TcpConnection::connect_established() {
    set_state(kConnected);
    channel_->tie(shared_from_this());
    channel_->enable_reading(); // 向poller注册channel的epollin事件

    // 新连接建立，执行回调
    connection_callback_(shared_from_this());
}

void TcpConnection::connect_destroyed() {
    if (state_ == kConnected) {  //渐进式转换，防止在关闭的时候没有输出完成而没有关的保底选项
        set_state(kDisconnected);
        channel_->disable_all(); // 把channel的所有感兴趣的事件，从poller中del掉
        connection_callback_(shared_from_this());
    }
    channel_->remove(); // 把channel从poller中删除掉
}
